/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.mqSender;

import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import static cn.ac.iie.mqSender.DistriThread.msgQueue;
import cn.ac.iie.unpack.file.global.GlobalParas;
import cn.ac.iie.unpack.file.main.UnpackFileMain;
import cn.ac.iie.unpack.file.utils.FileUntarAndMove;
import cn.ac.iie.unpack.file.utils.FileUntarAndMoveThread;
import com.alibaba.fastjson.JSONObject;
import java.io.File;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import jdk.nashorn.internal.objects.Global;
import jdk.nashorn.internal.objects.NativeArray;
import org.apache.avro.generic.GenericData;
import org.apache.log4j.Logger;
import org.apache.commons.configuration2.ex.ConfigurationException;
import org.apache.commons.lang.exception.ExceptionUtils;

/**
 *
 * @author gjw
 */
public class SendMQThread implements Callable<Integer>{

    static Logger LOG = Logger.getLogger(SendMQThread.class);
    
    String msgType;
    SendMQ sendMQ;
    
    ArrayBlockingQueue<List<JSONObject>> msgQueue=new ArrayBlockingQueue<List<JSONObject>>(100);
    long startTime ;
    String[] columns;
    String[] types;
    String  dir;
    List<Integer> urls=new ArrayList<Integer>();
    List<JSONObject> dataList=new ArrayList<JSONObject>();
    FileUntarAndMove fileUntar = new FileUntarAndMove();
    public SendMQThread(String topic,String msgType,String[] columns,String[] types,String[] urls,String dir) {
        this.msgType = msgType;
        this.columns = columns;
        this.types = types;
        this.dir=dir;
        this.sendMQ =new SendMQ(columns,getRetype(types),topic);
        startTime = System.currentTimeMillis() / 1000;
        for(String url:urls){
            for(int i=0;i<this.columns.length;i++){
                if(url.equals(columns[i])){
                    this.urls.add(i);
                }
            }
        }
    }
    public void add(JSONObject data) { 
        this.dataList.add(data);
        long endTime = System.currentTimeMillis() / 1000;
        if(dataList.size()>=GlobalParas.SEND_MQ_MSG_NUM||endTime-startTime>GlobalParas.SEND_MQ_TIME_OUT){
            try {
                msgQueue.put(dataList);
            } catch (InterruptedException ex) {
                LOG.error(ExceptionUtils.getFullStackTrace(ex));
            }
            dataList=new ArrayList<JSONObject>();
        }
        startTime=endTime;
    }
    @Override
    public Integer call() {
        List<JSONObject> jsonList = null;
        Thread.currentThread().setName(msgType);
        this.sendMQ.start();
        while (!DistriThread.Stop||!msgQueue.isEmpty()||!dataList.isEmpty()) {
            if(DistriThread.Stop&&!dataList.isEmpty()){
                try {
                    msgQueue.put(dataList);
                } catch (InterruptedException ex) {
                    LOG.error(ExceptionUtils.getFullStackTrace(ex));
                }
                dataList=new ArrayList<JSONObject>();
            }
            try {
                jsonList = msgQueue.poll(30, TimeUnit.SECONDS);
            } catch (InterruptedException ex) {
                LOG.error(ExceptionUtils.getFullStackTrace(ex));
            }
            if(jsonList==null){
                if(DistriThread.Stop){
                    LOG.info("SendMQThread msgQueue isEmpty to stop"+msgType);
                    long endTime = System.currentTimeMillis() / 1000;
                }
                LOG.info("SendMQThread msgQueue isEmpty");
                continue;
            }
            try {
                List<Object[]>datalist=new ArrayList<Object[]>();
                for(JSONObject json:jsonList){
                    Object[] data=new Object[columns.length];
                    for(int i=0;i<columns.length;i++){
                        if("strings".equals(types[i])){
                            data[i]=json.getJSONArray(columns[i]).toArray(new String[json.getJSONArray(columns[i]).size()]);
                        }else{
                            data[i]=json.getString(columns[i]);
                        }
                    }
                    for(int i:urls){
                        try{
                            for(String path:(String[])data[i]){
                                if(new File(GlobalParas.PARSER_DESTINATION_DIR+path).exists()){
                                   fileUntar.movefile(new File(GlobalParas.PARSER_DESTINATION_DIR+path),dir+path);
                                }else if(new File(dir+path).exists()){
                                }else{
                                   LOG.info("can_not_find "+path);
                                };
                            }
                        }catch(ClassCastException ex){
                            if(new File(GlobalParas.PARSER_DESTINATION_DIR+data[i]).exists()){
                                fileUntar.movefile(new File(GlobalParas.PARSER_DESTINATION_DIR+data[i]),dir+data[i]);
                            }else if(new File(dir+data[i]).exists()){
                            }else{
                                LOG.info("can_not_find "+data[i]);
                            };
                        }
                    }
                    datalist.add(data);
                }
                this.sendMQ.sendMessage(datalist);
                LOG.info("SendMQThread Number "+jsonList.size());
            } catch (Exception e) {
                LOG.error(ExceptionUtils.getFullStackTrace(e));
            }
            
        }
        this.sendMQ.stop();
        return 1;
    }
    private REFieldType[] getRetype(String[] types) {
        REFieldType[]retypes=new REFieldType[types.length];
        for(int i=0;i<types.length;i++){
            switch (types[i]) {
                case "int":
                    retypes[i]=REFieldType.Int;
                    break;
                case "long":
                    retypes[i]=REFieldType.Long;
                    break;
                case "string":
                    retypes[i]=REFieldType.String;
                    break;
                case "strings":
                    retypes[i]=REFieldType.Strings;
                    break;
                default:
                    LOG.error("There is not the type of " + types[i]);
            }
        }
        return retypes;
    }

    @Override
    public String toString() {
        return msgType;
    }
    

}
